package com.atguigu.flink.chapter05.Source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;
import java.util.Random;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 10:26
 */
public class Flink04_Source_MySource {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 自定义Source
        DataStreamSource<WaterSensor> sensorDS = env.addSource(new MySourceFunction());

        sensorDS.print();

        env.execute();
    }

    public static class MySourceFunction implements SourceFunction<WaterSensor> {
        // volatile : Java的内容，保证 可见性
        private volatile boolean isRunning = true;

        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {

            Random random = new Random();
            while (isRunning) {
                ctx.collect(
                        new WaterSensor(
                                "sensor_" + random.nextInt(3),
                                System.currentTimeMillis(),
                                random.nextInt(10))
                );
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
/**
 *  自定义 SourceFunction：
 *      1. 实现 SourceFunction相关接口
 *      2. 重写两个方法：
 *              run(): 主要逻辑
 *              cancel(): 停止逻辑
 *
 *   如果希望 Source可以指定并行度，那么就 实现 ParallelSourceFunction 这个接口
 */
